package com.swak.reactivex.transport;

import static java.util.concurrent.TimeUnit.MILLISECONDS;

import java.io.File;
import java.io.FileNotFoundException;
import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.OS;
import com.swak.exception.RemotingException;
import com.swak.reactivex.transport.codec.MultipleCodec;
import com.swak.reactivex.transport.resources.LoopResources;
import com.swak.registry.URL;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.timeout.IdleStateHandler;

/**
 * use Exchangers.connect();
 * 
 * @author 618lf
 */
public class NettyClient extends Endpoint implements Client {

	protected Logger logger = LoggerFactory.getLogger(NettyClient.class);

	private volatile Channel channel;
	private Bootstrap bootstrap;
	private final LoopResources resources;
	private final Lock connectLock = new ReentrantLock();

	public NettyClient(URL url, LoopResources resources, ChannelHandler handler) throws RemotingException {
		super(url, handler);
		this.resources = resources;

		try {
			open();
		} catch (Throwable t) {
			close();
			throw new RemotingException("Failed to start " + getClass().getSimpleName() + " connect to the server "
					+ remoteAddress() + ", cause: " + t.getMessage(), t);
		}

		try {
			// connect.
			connect();
			if (logger.isInfoEnabled()) {
				logger.info("Start " + getClass().getSimpleName() + " " + OS.localAddress() + " connect to the server "
						+ remoteAddress());
			}
		} catch (RemotingException t) {
			close();
			throw t;
		} catch (Throwable t) {
			close();
			throw new RemotingException("Failed to start " + getClass().getSimpleName() + " " + OS.localAddress()
					+ " connect to the server " + remoteAddress() + ", cause: " + t.getMessage(), t);
		}

	}

	private void open() {
		bootstrap = new Bootstrap();
		bootstrap.option(ChannelOption.SO_KEEPALIVE, true)//
				.option(ChannelOption.TCP_NODELAY, true) //
				.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)//
				.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, Math.max(3000, this.getConnectionTimeout()));//

		// 配置： Bootstrap
		this.customBootstrap(bootstrap);

		bootstrap.handler(new ChannelInitializer<SocketChannel>() {
			@Override
			public void initChannel(SocketChannel ch) throws Exception {
				ChannelPipeline pipeline = ch.pipeline();
                // 出方向  从下往上; 入方向 从上往下
				// 配置： SocketChannel
				NettyClient.this.customChannel(ch);
				pipeline.addLast("client-idle", new IdleStateHandler(getHeartbeat(), 0, 0, MILLISECONDS));
				pipeline.addLast("handler", new NettyClientHandler(getUrl(), NettyClient.this));
			}
		});
	}

	@Override
	public void connect() throws RemotingException {
		connectLock.lock();
		try {
			if (isConnected()) {
				return;
			}

			if (isClosed() || isClosing()) {
				logger.warn("No need to connect to server " + remoteAddress() + " from " + getClass().getSimpleName()
						+ " " + OS.ip() + ", cause: client status is closed or closing.");
				return;
			}

			doConnect();

			if (!isConnected()) {
				throw new RemotingException(
						"Failed connect to server " + remoteAddress() + " from " + getClass().getSimpleName() + " "
								+ OS.ip() + ", cause: Connect wait timeout: " + getConnectionTimeout() + "ms.");

			} else {
				if (logger.isInfoEnabled()) {
					logger.info("Successed connect to server " + remoteAddress() + " from " + getClass().getSimpleName()
							+ " " + OS.ip() + ", channel is " + this.getChannel());
				}
			}

		} catch (RemotingException e) {
			throw e;

		} catch (Throwable e) {
			throw new RemotingException("Failed connect to server " + remoteAddress() + " from "
					+ getClass().getSimpleName() + " " + OS.ip() + ", cause: " + e.getMessage(), e);

		} finally {
			connectLock.unlock();
		}
	}

	private void doConnect() throws RemotingException {
		long start = System.currentTimeMillis();
		ChannelFuture future = bootstrap.connect(this.getConnectAddress());
		boolean ret = future.awaitUninterruptibly(this.getConnectionTimeout(), TimeUnit.MILLISECONDS);
		if (ret && future.isSuccess()) {
			Channel newChannel = future.channel();
			try {
				Channel oldChannel = NettyClient.this.channel;
				if (oldChannel != null) {
					if (logger.isInfoEnabled()) {
						logger.info(
								"Close old netty channel " + oldChannel + " on create new netty channel " + newChannel);
					}
					oldChannel.close();
				}
			} finally {
				if (NettyClient.this.isClosed()) {
					try {
						if (logger.isInfoEnabled()) {
							logger.info("Close new netty channel " + newChannel + ", because the client closed.");
						}
						newChannel.close();
					} finally {
						NettyClient.this.channel = null;
					}
				} else {
					NettyClient.this.channel = newChannel;
				}
			}
		} else if (future.cause() != null) {
			throw new RemotingException("client failed to connect to server " + remoteAddress() + ", error message is:"
					+ future.cause().getMessage(), future.cause());
		} else {
			throw new RemotingException("client failed to connect to server " + remoteAddress()
					+ " client-side timeout " + this.getConnectionTimeout() + "ms (elapsed: "
					+ (System.currentTimeMillis() - start) + "ms) from netty client ");
		}
	}

	private InetSocketAddress getConnectAddress() {
		return new InetSocketAddress(getUrl().getHost(), getUrl().getPort());
	}

	@Override
	public void close() {
		if (isClosed()) {
			logger.warn("No need to close connection to server " + remoteAddress() + " from "
					+ getClass().getSimpleName() + " " + OS.ip() + ", cause: the client status is closed.");
			return;
		}

		connectLock.lock();
		try {
			if (isClosed()) {
				logger.warn("No need to close connection to server " + remoteAddress() + " from "
						+ getClass().getSimpleName() + " " + OS.ip() + ", cause: the client status is closed.");
				return;
			}

			try {
				super.close();
			} catch (Throwable e) {
				logger.warn(e.getMessage(), e);
			}

			try {
				disconnect();
			} catch (Throwable e) {
				logger.warn(e.getMessage(), e);
			}

		} finally {
			connectLock.unlock();
		}
	}

	public void disconnect() {
		connectLock.lock();
		try {
			try {
				com.swak.reactivex.transport.Channel channel = getChannel();
				if (channel != null) {
					channel.close();
				}
			} catch (Throwable e) {
				logger.warn(e.getMessage(), e);
			}
			try {
				NettyChannel.removeChannelIfDisconnected(channel);
			} catch (Throwable e) {
				logger.warn(e.getMessage(), e);
			}
		} finally {
			connectLock.unlock();
		}
	}

	@Override
	public void send(Object message, boolean sent) throws RemotingException {
		if (!isConnected()) {
			connect();
		}
		com.swak.reactivex.transport.Channel channel = getChannel();
		if (channel == null || !channel.isConnected()) {
			throw new RemotingException("message can not send, because channel is closed . url:" + getUrl());
		}
		channel.send(message, sent);
	}

	@Override
	public void sendFile(File file, boolean sent) throws RemotingException, FileNotFoundException {
		if (!isConnected()) {
			connect();
		}
		com.swak.reactivex.transport.Channel channel = getChannel();
		if (channel == null || !channel.isConnected()) {
			throw new RemotingException("message can not send, because channel is closed . url:" + getUrl());
		}
		channel.sendFile(file, sent);
	}

	@Override
	public boolean isConnected() {
		com.swak.reactivex.transport.Channel channel = getChannel();
		if (channel == null) {
			return false;
		}
		return channel.isConnected();
	}

	@Override
	public InetSocketAddress localAddress() {
		com.swak.reactivex.transport.Channel channel = getChannel();
		if (channel == null) {
			return InetSocketAddress.createUnresolved(OS.ip(), 0);
		}
		return channel.localAddress();
	}

	@Override
	public InetSocketAddress remoteAddress() {
		com.swak.reactivex.transport.Channel channel = getChannel();
		if (channel == null) {
			return getUrl().toInetSocketAddress();
		}
		return channel.remoteAddress();
	}

	private com.swak.reactivex.transport.Channel getChannel() {
		Channel c = channel;
		if (c == null) {
			return null;
		}
		return NettyChannel.getOrAddChannel(getUrl(), c, this);
	}

	@Override
	public Object getAttribute(String key) {
		com.swak.reactivex.transport.Channel channel = getChannel();
		if (channel == null) {
			return null;
		}
		return channel.getAttribute(key);
	}

	@Override
	public void setAttribute(String key, Object value) {
		com.swak.reactivex.transport.Channel channel = getChannel();
		if (channel == null) {
			return;
		}
		channel.setAttribute(key, value);
	}

	@Override
	public void removeAttribute(String key) {
		com.swak.reactivex.transport.Channel channel = getChannel();
		if (channel == null) {
			return;
		}
		channel.removeAttribute(key);
	}

	@Override
	public boolean hasAttribute(String key) {
		com.swak.reactivex.transport.Channel channel = getChannel();
		if (channel == null) {
			return false;
		}
		return channel.hasAttribute(key);
	}

	/**
	 * 配置编解码协议
	 * 
	 * @param ch
	 */
	protected void customChannel(SocketChannel ch) {
		ChannelPipeline pipeline = ch.pipeline();
		Codec codec = new MultipleCodec(getUrl());
		NettyCodec nettyCodec = new NettyCodec(codec, getUrl(), this);
		pipeline.addLast(NettyCodec.Encoder_Name, nettyCodec.getEncoder());
		pipeline.addLast(NettyCodec.Decoder_Name, nettyCodec.getDecoder());
	}

	/**
	 * 配置事件组，配置各种参数
	 * 
	 * @param bootstrap
	 */
	protected void customBootstrap(Bootstrap bootstrap) {
		bootstrap.group(resources.onClient()).channel(resources.onClientChannel());
	}
}